package ex.tableapi.aggregation;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.*;

public class ExDistinctAggregation extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExDistinctAggregation demo = new ExDistinctAggregation();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table result = demo.exDistinctAggregation(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }



    private Table exDistinctAggregation(Table orders) {
        Table groupByDistinctResult = orders
                .groupBy($("a"))
                .select($("a"), $("b").sum().distinct().as("d"));
        //按a,rowtime(5分钟一次or 1小时)分组，对b去重求和
        Table groupByWindowDistinctResult = orders
                .window(Tumble
//                        .over(lit(5).minutes())
                                .over(lit(1).hour())//测试数据在小时区间，b列重复金额会去重
                                .on($("rowtime"))
                                .as("w")
                )
                .groupBy($("a"), $("w"))
                .select($("a"), $("b").sum().distinct().as("d"));
// Distinct aggregation on over window

        Table result = orders
                .window(Over
                        .partitionBy($("a"))
                        .orderBy($("rowtime"))
                        .preceding(UNBOUNDED_RANGE)
                        .as("w"))
                .select(
                        $("a"),
                        $("b").avg().distinct().over($("w")),
                        $("b").max().over($("w")),
                        $("b").min().over($("w"))
                );
//        Table result = orders.distinct();
        return groupByWindowDistinctResult;
    }
}